package com.inventive.system.monitoring.server.service.publisher;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IMap;
import com.hazelcast.core.Member;
import com.hazelcast.core.MultiTask;
import com.hazelcast.query.EntryObject;
import com.hazelcast.query.Predicate;
import com.hazelcast.query.PredicateBuilder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Required;

import javax.annotation.PostConstruct;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class InProcessStatisticPublisher implements MapNames, StatisticPublisher {

    private static final Log log = LogFactory.getLog(InProcessStatisticPublisher.class);

    @Autowired(required = true)
    private MBeanServer mbs;

    private HazelcastInstance hazelcastInstance;

    private List<StatisticDescriptor> statistics;

    private Set<StatisticDescriptor> locks = new HashSet<StatisticDescriptor>();

    private IMap<StatisticKey, StatisticEntry> statisticsMap;

    private String machineName;

    private String processName;

    private ScheduledExecutorService executorService;

    @PostConstruct
    private void init() throws Exception {

        machineName = getMachineName();
        statisticsMap = hazelcastInstance.getMap(STATISTICS_MAP);
        executorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors() * 2);

        //Try and acquire locks for machine statistics
        executorService.scheduleAtFixedRate(new Runnable() {

            public void run() {

                if (null != statistics && statistics.size() > 0 && locks.size() < statistics.size()) {
                    IMap<MachineLockKey, Integer> map = hazelcastInstance.getMap(MACHINE_LOCKS);

                    for (StatisticDescriptor descriptor : statistics) {
                        if (!locks.contains(descriptor)) {
                            boolean result = map.tryLock(new MachineLockKey(machineName, descriptor), 500L, TimeUnit.MILLISECONDS);
                            if (result) {
                                locks.add(descriptor);
                            }
                        }
                    }
                }


            }
        }, 5L, 30L, TimeUnit.SECONDS);

        for (final StatisticDescriptor descriptor : statistics) {
            executorService.scheduleAtFixedRate(new RecordStatistic(descriptor), descriptor.getInterval(), descriptor.getInterval(), TimeUnit.MILLISECONDS);
        }

        executorService.scheduleAtFixedRate(new RemoveRedundantEntriesRunnable(), 10L, 10L, TimeUnit.SECONDS);


    }

    private String getMachineName() {
        if (null == machineName) {
            try {
                machineName = InetAddress.getLocalHost().getHostName();
            } catch (UnknownHostException e) {
                log.error(e);
                return null;
            }
        }
        return machineName;
    }

    private class RemoveRedundantEntriesRunnable implements Runnable {
        public void run() {
            Set<Member> members = hazelcastInstance.getCluster().getMembers();

            MultiTask<Void> task = new MultiTask<Void>(new Callable<Void>() {
                public Void call() throws Exception {


                    EntryObject e = new PredicateBuilder().getEntryObject();
                    Predicate predicate = e.lessThan(new Date());

                    IMap<StatisticKey, StatisticEntry> processStatisticsMap = hazelcastInstance.getMap(STATISTICS_MAP);
                    Set<Map.Entry<StatisticKey, StatisticEntry>> processEntries = processStatisticsMap.entrySet(predicate);

                    for (Map.Entry<StatisticKey, StatisticEntry> entry : processEntries) {
                        processStatisticsMap.remove(entry.getKey());
                    }

                    return null;
                }
            }, members);
            hazelcastInstance.getExecutorService().execute(task);
        }
    }

    @Qualifier("statistics")
    public void setStatistics(List<StatisticDescriptor> statistics) {
        this.statistics = statistics;
    }

    @Required
    public void setHazelcastInstance(HazelcastInstance hazelcastInstance) {
        this.hazelcastInstance = hazelcastInstance;
    }

    @SuppressWarnings("UnusedDeclaration")
    public void setMachineName(String machineName) {
        this.machineName = machineName;
    }

    private class RecordStatistic implements Runnable {
        private final StatisticDescriptor descriptor;

        public RecordStatistic(StatisticDescriptor descriptor) {
            this.descriptor = descriptor;
        }

        @SuppressWarnings("unchecked")
        public void run() {
            ObjectName objectName = null;
            try {
                objectName = new ObjectName(descriptor.getObjectName());
            } catch (Exception e) {
                log.error(e);
            }

            try {

                Object value = mbs.getAttribute(objectName, descriptor.getPropertyName());

                if (descriptor.getConverter() != null) {
                    value = descriptor.getConverter().convert(value);
                }

                Date date = new Date();

                statisticsMap.put(new StatisticKey(machineName, processName, date, descriptor.getObjectName(), descriptor.getPropertyName(), descriptor.getFieldName()), new StatisticEntry(value, date, getMachineName(), processName, descriptor.getObjectName(), descriptor.getPropertyName(), descriptor.getFieldName()));

            } catch (Exception e) {
                log.error(e);
            }
        }
    }

    @Override
    public void addDescriptor(StatisticDescriptor descriptor) {
        ObjectName objectName;
        try {
            objectName = new ObjectName(descriptor.getObjectName());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        try {

            Object value = mbs.getAttribute(objectName, descriptor.getPropertyName());
            if (null != value) {
                executorService.scheduleAtFixedRate(new RecordStatistic(descriptor), descriptor.getInterval(), descriptor.getInterval(), TimeUnit.MILLISECONDS);
            }


        } catch (Exception e) {
            log.error(e);
        }
    }

    public void setProcessName(String processName) {
        this.processName = processName;
    }
}
